# include <sys/socket.h>
# include <netinet/in.h>
# include <stdio.h>
# include <netdb.h>
# include <string.h>
# include <sys/types.h>
# include <unistd.h>
# include <math.h>
# include <stdlib.h>
# include <sys/stat.h>
# include <errno.h>
# include <fcntl.h>
# include <sys/time.h>

# include <pthread.h>

# include <vector>
# include <queue>
# include <string>
# include <locale>
# include <set>
# include <functional>

using namespace std;

# define MYTCP_PORT 31038
# define REMOTE_PORT 80

# define MAXTHREAD 1

# define BACKLOG 10
# define MAXSIZE 250

typedef unsigned long long ull;

struct URL
{
	string url;
	string domain;
	string path;

	double responseTime;
};

struct ThreadParam
{
	int id;
	URL url;
};

bool runningParser;
bool occupy;

pthread_t worker, querier;
pthread_mutex_t threadStatusMutex;
pthread_mutex_t queueMutex;
pthread_mutex_t urlSetMutex;
pthread_mutex_t domainSetMutex;
pthread_mutex_t m_SuspendMutex;
pthread_cond_t m_ResumeCond;

pthread_t ioThread, parseThread;
pthread_mutex_t lineMutex;
bool finish;

clock_t start, pfinish;

deque<string> line;

int m_SuspendFlag;
int crawlerStatus;

int sockfd, con_fd, ret;
struct sockaddr_in my_addr;
struct sockaddr_in their_addr;
socklen_t sin_size;

pid_t pid;

queue<URL> url;

pthread_t thread[MAXTHREAD];
time_t threadRunningTime[MAXTHREAD];
bool threadStatus[MAXTHREAD];

hash<string> stringHash;
set<ull> urlSet;
set<ull> domainSet;

string curLine;

struct timeval sendPacket, recvPacket;

// templated version of my_equal so it could work with both char and wchar_t
template<typename charT>
struct my_equal {
	my_equal( const std::locale& loc ) : loc_(loc) {}
	bool operator()(charT ch1, charT ch2) {
		return std::toupper(ch1, loc_) == std::toupper(ch2, loc_);
	}
private:
	const std::locale& loc_;
};

// find substring (case insensitive)
int ci_find_substr( const string& str1, const string& str2, const std::locale& loc = std::locale() )
{
	string::const_iterator it = std::search( str1.begin(), str1.end(), 
		str2.begin(), str2.end(), my_equal<string::value_type>(loc) );
	if ( it != str1.end() ) return it - str1.begin();
	else return -1; // not found
}

double getResponseTime(struct  timeval *out, struct timeval *in)
{
	double time_invt;
	printf("Calculating response time\n");
	if ((out->tv_usec -= in->tv_usec) <0)
	{
		--out ->tv_sec;
		out ->tv_usec += 1000000;
	}
	out->tv_sec -= in->tv_sec;

	time_invt = (out->tv_sec)*1000.0 + (out->tv_usec)/1000.0;

	printf("Ending calculating response time\n");

	return time_invt;
}

bool getByte(int socket, char &x)
{
	int n;

	if ((n = recv(socket, &x, 1, 0))==-1)
	{
		printf("error when receiving\n");
		return false;
	}

	return true;
}

bool getString(int socket, string &s)
{
	char newByte = 1;
	s = "";

	while (newByte)
	{
		if (!getByte(socket, newByte))
		{
			printf("error when receiving\n");
			return false;
		}

		s += newByte;
	}

	return true;
}

void clearQueue(queue<URL> &q)
{
	while (!q.empty())
	{
		q.pop();
	}
}

void suspendThreads()
{
	pthread_mutex_lock(&m_SuspendMutex);
	m_SuspendFlag = 1;
	pthread_mutex_unlock(&m_SuspendMutex);
}

void resumeThreads()
{
	pthread_mutex_lock(&m_SuspendMutex);
	m_SuspendFlag = 0;
	pthread_cond_broadcast(&m_ResumeCond);
	pthread_mutex_unlock(&m_SuspendMutex);
}

void checkSuspend()
{
	pthread_mutex_lock(&m_SuspendMutex);
	while (m_SuspendFlag != 0) pthread_cond_wait(&m_ResumeCond, &m_SuspendMutex);
	pthread_mutex_unlock(&m_SuspendMutex);
}

bool isWhiteSpace(char x)
{
	return x == ' ' || x == '\t' || x == '\n' || x == '\r';
}

bool duplicateURL(string url)
{
	if (!urlSet.empty() && urlSet.count(stringHash(url)))
	{
		pthread_mutex_unlock(&urlSetMutex);
		return true;
	}

	// push the current url into the set
	urlSet.insert(stringHash(url));
	pthread_mutex_unlock(&urlSetMutex);
	return false;
}

bool duplicateDomain(string domain)
{
	if (!domainSet.empty() && domainSet.count(stringHash(domain)))
	{
		pthread_mutex_unlock(&domainSetMutex);
		return true;
	}

	// push the current url into the set
	domainSet.insert(stringHash(domain));
	pthread_mutex_unlock(&domainSetMutex);
	return false;
}

bool ignoreURL(URL url)
{
	string path = url.path;

	return path.find(".css") != string::npos || path.find(".js") != string::npos ||
		   path.find(".jsp") != string::npos || path.find(".php") != string::npos ||
		   path.find(".asp") != string::npos || path.find(".aspx") != string::npos;
}

URL parseURL(string def, string s, bool href)
{
	URL result;

	result.url = s;

	if (s.find("https://") == 0)
	{
		size_t pathPos = s.find("/", 8);

		if (pathPos != string::npos)
		{
			result.domain = s.substr(8, pathPos - 8);
			result.path = s.substr(pathPos);
		} else
		{
			result.domain = s.substr(8);
			result.path = "/";
		}
	} else
	if (s.find("http://") == 0)
	{
		size_t pathPos = s.find("/", 7);

		if (pathPos != string::npos)
		{
			result.domain = s.substr(7, pathPos - 7);
			result.path = s.substr(pathPos);
		} else
		{
			result.domain = s.substr(7);
			result.path = "/";
		}
	} else
	if (s.find("//") == 0)
	{
		size_t pathPos = s.find("/", 3);

		if (pathPos != string::npos)
		{
			result.domain = s.substr(3, pathPos - 3);
			result.path = s.substr(pathPos);
		} else
		{
			result.domain = s.substr(3);
			result.path = "/";
		}
	} else
	// ignore other protocals
	if (s.find(":") == string::npos)
	{
		if (!href)
		{
			if (s.find("/") != string::npos)
			{
				int pos = s.find("/");

				result.domain = s.substr(0, pos);
				result.path = s.substr(pos);
			} else
			{
				result.domain = s;
				result.path = "/";
			}
		} else
		{
			if (s[0] == '/')
			{
				result.domain = def;
				result.path = s;
			} else
			{
				result.domain = def;
				result.path = "/" + s;
			}
		}
	}

	if (result.path == "//") result.path = "/";

	return result;
}

bool readLineFromSocket(int socket, bool &fin, string &curLine)
{
	char buf[500];
	int n;

	// introduce 1ms delay to avoid consumer overwhelming
	usleep(5000);

	if ((n = recv(socket, buf, 491, 0))==-1)
	{
		printf("read file: error when receiving\n");
		return false;
	}

	printf("From reader: %d bytes\n", strlen(buf));

	fin = (strlen(buf) < 491);
	buf[strlen(buf)] = 0;
	string str(buf);
	
	if (str.find("</html>") != string::npos || str.find("</body>") != string::npos)
	{
		fin = true;
	}

	// printf("%s", str.c_str());
	str.c_str();
	curLine = str;

	return true;
}

void* readLine(void* param)
{
	bool fin = false;
	int socket = *(int*)param;
	string curLine;

	line.clear();

	while (!finish)
	{
		if (readLineFromSocket(socket, fin, curLine))
		{
			pthread_mutex_lock(&lineMutex);
			line.push_back(curLine);
			pthread_mutex_unlock(&lineMutex);

			finish = fin;
		} else
		{
			printf("Read Line error\n");
			return (void*) 1;
		}
	}

	// printf("Read Line succ\n");
	return (void*) 0;
}

bool getNextLine(string &next)
{
	while (!finish && line.empty());

	if (line.empty())
	{
		return false;
	}

	next = line.front();
	pthread_mutex_lock(&lineMutex);
	line.pop_front();
	pthread_mutex_unlock(&lineMutex);

	return true;
}

void* parseLine(void* param)
{
	string curLine, curUrl, nextLine;
	string pattern = " href";
	string patternLocation = "location:";
	
	int initState, matchState;
	int len, lenP, i, j;

	bool match, curLineFinsh, header;

	URL parsedURL;

	string defDomain = *(string*)param;

	lenP = pattern.length();
	initState = matchState = 0;
	curLineFinsh = false;
	header = true;

	// when the reading haven't finished or the line queue
	// has not been emptied or current line parsing hasn't
	// finished
	try
	{
		while (!finish || !line.empty() || !curLineFinsh)
		{
			// printf("1\n");
			switch (initState)
			{
				// init state
				case 0:
					i = j = 0;
					match = true;
					curLineFinsh = false;
					if (getNextLine(nextLine))
					{
						curLine = nextLine;
					} else
					{
						return (void *) 1;
					}
					
					// look for location information first
					if (header)
					{
						// do find location in header only once
						header = false;

						int locationPos = ci_find_substr(curLine, patternLocation);

						// if we are able to find location information in the header
						if (locationPos != -1)
						{
							locationPos += 10;
							while (curLine[locationPos] == ' ') locationPos ++;
							int term = curLine.find("\n", locationPos);
							curUrl = curLine.substr(locationPos, term - locationPos);
							parsedURL = parseURL(defDomain, curUrl, true);
							
							// if we have encountered the same url before, ignore and continue
							pthread_mutex_lock(&urlSetMutex);
							if (!duplicateURL(parsedURL.domain + parsedURL.path))
							{
								// if (!ignoreURL(parsedURL))
								if (parsedURL.path.length() < 400)
								{
									url.push(parsedURL);
								}
							}

							curLine = curLine.substr(term + 1);
						}
					}

					len = curLine.length();
					curUrl = "";
					break;
				// read one more line state
				case 1:
					// printf("1.1, %d\n", curLine.length());
					if (getNextLine(nextLine))
					{
						curLine += nextLine;
					} else
					{
						return (void *) 1;
					}
					// printf("1.1.1\n");
					len = curLine.length();
					break;
				// continue finding " href" state
				case 2:
					// printf("1.2\n");
					i ++;
					j = 0;
					match = true;
					matchState = 0;
				break;
			}

			// printf("2\n");
			checkSuspend();

			switch (matchState)
			{
				// finding <a href
				case 0:
					// string matching process, can be substitute with KMP to speed up
					// currently we haven't handled cases like <a [multiple spaces] href
					while (i + j < len && j < lenP)
					{
						if (tolower(curLine[i + j]) != pattern[j])
						{
							match = false;
							break;
						}

						j ++;
					}

					// if we've run out of current line
					if (i >= len)
					{
						// indicate finish of current line
						curLineFinsh = true;
						// get a new line
						initState = 0;
						break;
					}

					// if " href" is not found
					if (!match)
					{
						// continue matching
						initState = 2;
						break;
					}

					// if " href" is not found yet but still matches
					if (i + j >= len)
					{
						// read one more line and continue find <a href
						initState = 1;
						break;
					}

					// " href" is found, proceed to url matching
					i += lenP;
				// finding '='
				case 1:
					while (i < len && isWhiteSpace(curLine[i]))
					{
						i ++;
					}

					if (i >= len)
					{
						// read one more line and continue finding '='
						initState = 1;
						matchState = 1;
						break;
					} else
					if (curLine[i] != '=')
					{
						// continue matching
						initState = 2;
						break;
					} else
					{
						i ++;
					}
				// finding url start, i.e. left quotation mark ('"')
				case 2:
					while (i < len && isWhiteSpace(curLine[i]))
					{
						i ++;
					}

					if (i >= len)
					{
						// read one more line and continue finding '"'
						initState = 1;
						matchState = 2;
						break;
					} else
					if (curLine[i] != '\"')
					{
						// continue matching
						initState = 2;
						break;
					} else
					{
						i ++;
					}
				// finding url pfinish, i.e. right quotation mark ('"')
				case 3:
					while (i < len && curLine[i] != '\"')
					{
						if (!isWhiteSpace(curLine[i]))
						{
							curUrl += curLine[i];
						}

						i ++;
					}

					if (i >= len)
					{
						// read one more line and continue finding '"'
						initState = 1;
						matchState = 3;
						break;
					} else
					{
						// now we've finished parsing a string, back to init state
						initState = 2;

						parsedURL = parseURL(defDomain, curUrl, true);

						pthread_mutex_lock(&urlSetMutex);
						if (!duplicateURL(parsedURL.domain + parsedURL.path))
						{
							// if (!ignoreURL(parsedURL))
							if (parsedURL.path.length() < 400)
							{
								url.push(parsedURL);
							}
							// printf("%s\n", parsedURL.url.c_str());
						}

						curUrl = "";
						curLineFinsh = true;
						break;
					}
			}
		}
	} catch (...)
	{
		printf("fucking error wocaonima\n");
	}

	pthread_exit(NULL);	
}

bool parallelParser(string defDomain, int sockfd)
{
	void *ioStatus;

	runningParser = true;
	printf("start parallel parsing\n");

	pthread_mutex_init(&lineMutex, NULL);

	finish = false;

	// only for passing the integer into the thread

	pthread_create(&ioThread, NULL, readLine, (void *)&sockfd);
	pthread_create(&parseThread, NULL, parseLine, (void *)&defDomain);

	printf("Parsing started...\n");

	pthread_join(ioThread, &ioStatus);
	
	if ((int*)(ioStatus) == 0)
	{
		pthread_join(parseThread, NULL);
	} else
	{
		pthread_cancel(parseThread);
	}

	printf("Parsing completed\n");

	if ((int*)(ioStatus) != 0)
	{
		printf("Error when parsing\n");
		pthread_mutex_destroy(&lineMutex);
		return false;
	}

	pthread_mutex_destroy(&lineMutex);

	return true;
}

bool httpGet(int sockfd, hostent* sh, string path, double &responseTime)
{
	char sends[MAXSIZE * 2];

	int n = 0;
	int httpResponseStatus;
	int err = 0;

	char x = 0;

	printf("Host Name: %s\n", sh->h_name);

	memset(sends, 0, sizeof(sends));
	sprintf(sends, "GET %s HTTP/1.1\nHost: %s\n\n", path.c_str(), sh->h_name);
	
	printf("%s\n", sends);

	gettimeofday(&sendPacket,NULL);
	send(sockfd, sends, strlen(sends), 0);		//send the string to the server(receiver)

	// polling for HTTP Response
	printf("Wait for response\n");
	bool success = getByte(sockfd, x);
	printf("%d\n", success);
	if (!success)
	{
		printf("Error when fetching\n");
		return false;
	}
	gettimeofday(&recvPacket,NULL);
	// fprintf(fp, "%c", x);
	// printf("%c", x);
	// printf("%d\n", success);

	responseTime = getResponseTime(&recvPacket, &sendPacket);
	
	printf("Response Time = %.4lfms\n", responseTime);

	// printf("%d\n", success);

	if (!success)
	{
		printf("Error when fetching\n");
		return false;
	}

	return parallelParser(string(sh->h_name), sockfd);
}

void *newRequest(void* param)
{
	int sockfd_http_tcp, ret, threadID;
	struct sockaddr_in ser_addr;
	struct in_addr **addrs;
	hostent* sh;

	URL urlInput = *(struct URL*) param;
	double responseTime;
	char sends[100];

	printf("Start new request\n");

	sh = gethostbyname(urlInput.domain.c_str());

	if (sh == NULL)
	{
		printf("Error when gethostby name\n");
		pthread_mutex_lock(&threadStatusMutex);
		occupy = false;
		pthread_mutex_unlock(&threadStatusMutex);
		return (void *) 1;
	}

	// routine below is to bind a socket with the TCP channel
	addrs = (struct in_addr **)sh->h_addr_list;
	sockfd_http_tcp = socket(AF_INET, SOCK_STREAM, 0);
	
	if (sockfd_http_tcp < 0)
	{
		printf("Error in socket");
		pthread_mutex_lock(&threadStatusMutex);
		occupy = false;
		pthread_mutex_unlock(&threadStatusMutex);
		return (void *) 1;
	}

	ser_addr.sin_family = AF_INET;
	ser_addr.sin_port = htons(REMOTE_PORT);
	memcpy(&(ser_addr.sin_addr.s_addr), *addrs, sizeof(struct in_addr));	
	bzero(&(ser_addr.sin_zero), 8);

	ret = connect(sockfd_http_tcp, (struct sockaddr *)&ser_addr, sizeof(struct sockaddr));
	if (ret != 0)
	{
		printf ("Connection failed\n"); 
		close(sockfd_http_tcp); 
		pthread_mutex_lock(&threadStatusMutex);
		occupy = false;
		pthread_mutex_unlock(&threadStatusMutex);
		return (void *) 1;
	}

	checkSuspend();
	
	// send http GET request to retrieve html, record response time in messenger
	if (httpGet(sockfd_http_tcp, sh, urlInput.path, responseTime))
	{
		runningParser = false;
		printf("Get done\n");

		if (!duplicateDomain(urlInput.domain))
		{
			// send the url and its response time to client
			sprintf(sends, "%s", urlInput.domain.c_str());
			sends[strlen(sends)] = 0;
			send(con_fd, sends, strlen(sends) + 1, 0);
			sprintf(sends, "%.3lf", responseTime);
			sends[strlen(sends)] = 0;
			printf("res = %s\n", sends);
			send(con_fd, sends, strlen(sends) + 1, 0);
		}
	} else
	{
		runningParser = false;
		printf("Get failed\n");
		pthread_mutex_lock(&threadStatusMutex);
		occupy = false;
		pthread_mutex_unlock(&threadStatusMutex);
		return (void *) 1;
	}

	pthread_mutex_lock(&threadStatusMutex);
	occupy = false;
	pthread_mutex_unlock(&threadStatusMutex);

	close(sockfd_http_tcp);
	return (void *) 0;
}

void *crawl(void * p)
{
	time_t start, end;

	printf("Crawler started\n");
	occupy = false;
	runningParser = false;

	while (true)
	{
		checkSuspend();

		// create the new task
		if (!occupy)
		{
			printf("Retrieving url, running parser = %d\n", runningParser);
			if (runningParser)
			{
				pthread_join(querier, NULL);
				pthread_join(ioThread, NULL);
				pthread_join(parseThread, NULL);
			}
			// retrieve current url information
			URL curUrl = url.front();
			url.pop();

			start = clock();
			printf("current = %s\n", curUrl.url.c_str());
			printf("path = %s\n", curUrl.path.c_str());
			pthread_mutex_lock(&threadStatusMutex);
			occupy = true;
			pthread_mutex_unlock(&threadStatusMutex);
			pthread_create(&querier, NULL, newRequest, (void *)&curUrl);
		} else
		{
			end = clock();

			if (!runningParser && end - start > 5 * CLOCKS_PER_SEC)
			{
				printf("Timeout, killing\n");
				pthread_cancel(querier);
				pthread_cancel(ioThread);
				pthread_cancel(parseThread);
				occupy = false;
			}
		}
	}

	pthread_exit(NULL);
}

// communicating server state machine
void serverSM(int sockfd)
{
	char opcode;
	char totalURL;
	char newByte;
	
	printf("receiving data!\n");

	while (true)
	{
		printf("at this fucking point\n");
		// get instruction
		getByte(sockfd, opcode);
		printf("Opcode = %d\n", opcode);

		// start with base URL
		if (opcode == 0 && getByte(sockfd, totalURL))
		{
			printf("Total URL to receive = %d\nBase URL:\n", totalURL);
			clearQueue(url);

			while(totalURL > 0)
			{
				string s;

				getString(sockfd, s);

				URL rawURL = parseURL("localhost", s + "/", false);
				
				pthread_mutex_lock(&urlSetMutex);
				if (!duplicateURL(rawURL.domain + rawURL.path))
				{
					if (!ignoreURL(rawURL))
					{
						url.push(rawURL);
						printf("%s\n", rawURL.url.c_str());
					}
				}

				totalURL --;
			}

			// check if we have a running crawler
			if (crawlerStatus)
			{
				printf("Killing running crawler\n");
				pthread_cancel(worker);
				pthread_cancel(ioThread);
				pthread_cancel(parseThread);
				crawlerStatus = 0;
			}

			// restart the crawler
			crawlerStatus = 1;
			pthread_create(&worker, NULL, crawl, NULL);
			
			printf("Receving Complete.\n");
		} else
		// start with old queue
		if (opcode == 1)
		{
			resumeThreads();
			printf("Crawler resumed\n");
			// resume the crawler
		} else
		// stop
		if (opcode == 2)
		{
			suspendThreads();
			printf("Crawler suspended\n");
			// suspend the crawler
		}
	}

	pthread_cancel(worker);
	printf("Worker done\n");
}

int main()
{
//	char *buf;

//------------------- Bind Current Application to a Socket ------------------//
	sockfd = socket(AF_INET, SOCK_STREAM, 0);	  //create socket
	if (sockfd <0)
	{
		printf("error in socket!");
		exit(1);
	}
	
	my_addr.sin_family = AF_INET;
	my_addr.sin_port = htons(MYTCP_PORT);
	my_addr.sin_addr.s_addr = htonl(INADDR_ANY);//inet_addr("172.0.0.1");
	bzero(&(my_addr.sin_zero), 8);
	ret = bind(sockfd, (struct sockaddr *) &my_addr, sizeof(struct sockaddr));	//bind socket
	if (ret <0)
	{
		printf("error in binding");
		exit(1);
	}
	
	ret = listen(sockfd, BACKLOG);	  //listen
	if (ret <0) {
		printf("error in listening");
		exit(1);
	}

//--------------------------------- Start Serving ------------------------------------------//
	crawlerStatus = 0;
	m_SuspendFlag = 0;

	pthread_mutex_init(&m_SuspendMutex, NULL);
	pthread_mutex_init(&queueMutex, NULL);
	pthread_mutex_init(&urlSetMutex, NULL);
	pthread_mutex_init(&domainSetMutex, NULL);
	pthread_mutex_init(&threadStatusMutex, NULL);
	pthread_cond_init(&m_ResumeCond, NULL);

	while (1)
	{
		printf("Waiting for URLs\n");
		sin_size = sizeof (struct sockaddr_in);
		con_fd = accept(sockfd, (struct sockaddr *)&their_addr, &sin_size);	//accept the packet
		if (con_fd <0)
		{
			printf("error in accept\n");
			exit(1);
		}

		if ((pid = fork())==0)	 // creat acception process
		{
			serverSM(con_fd);	  //receive packet and response
			close(con_fd);
			printf("Exiting\n");
			exit(0);
		}
		else close(con_fd);	 //parent process
	}

	printf("Program terminated\n");
	close(sockfd);
	exit(0);
}